package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionCommentInfo extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdInteractionCommentInfo().start(10012,4,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }



    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId)  {
        // 1.读取ods层数据
        readOdsDb(tEnv,groupId);

        // 2. 过滤评论数据
        Table comment = getcomment(tEnv);
        //创建临时表
        tEnv.createTemporaryView("comment_info",comment);

//        tEnv.sqlQuery("select * from comment_info").execute().print();

        // 3、读取码表 common
        //字典数据
        createBaseDic(tEnv);

        // 4、关联数据
        Table uniontable = uniontable(tEnv);

        // 5、写入到Kafka 创建 topic 表的数据结构
        extracted(tEnv);

        // 6写出去
        //将关联结果 插入 目标 topic内
        uniontable.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();

    }

    // 5、写入到Kafka
    // 现根据 要存入kafka的结构进行 创表 ，在次场景属于 创建 topic格式
    private static TableResult extracted(StreamTableEnvironment tEnv) {
        return tEnv.executeSql("create table " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO + "(" +
                "id string,\n" +
                "user_id string,\n" +
                "nick_name string ,\n" +
                "sku_id string ,\n" +
                "spu_id string ,\n" +
                "order_id string ,\n" +
                "appraise_code string ,\n" +
                "appraise_name string,\n" +
                "comment_txt string ,\n" +
                "create_time string \n" +
                ") "
                + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }

    //关联数据
    private static Table uniontable(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("SELECT \n" +
                " id,\n" +
                " user_id,\n" +
                " nick_name,\n" +
                " sku_id,\n" +
                " spu_id,\n" +
                " order_id,\n" +
                " appraise,\n" +
                " info.dic_name,\n" +
                " comment_txt,\n" +
                " create_time\n" +
                "FROM comment_info AS c\n" +
                "JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b\n" +
                "  ON c.appraise = b.rowkey");
    }

    //过滤评论数据
    private static  Table getcomment(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['nick_name'] nick_name,\n" +
                "`data`['head_img'] head_img,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['spu_id'] spu_id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['appraise'] appraise,\n" +
                "`data`['comment_txt'] comment_txt,\n" +
                "`data`['create_time'] create_time," +
                " proc_time \n" +
                "from topic_db\n" +
                "where `database`='gmall'  and `table` = 'comment_info'\n" +
                "and `type`='insert' ");
    }
}
